use std::sync::Arc;
use std::time::Duration;

use async_std::net::TcpStream;
use futures::channel::oneshot;
use xactor::{message, spawn, Actor, Addr, Context, Handler, Message};

use crate::actors::tcpstream::TcpStreamActor;
use crate::message::{
    MessageId, RpcResponseError, RpcResult, TcpRegister, UserMessageApp, UserMessageForward,
};
use crate::prelude::*;
use crate::Result;

pub struct TcpConnecterActor {
    actor_name: String,
    node_type: String,
    node_name: String,
    self_addr: Option<Addr<Self>>,
    remote_addr: String,
    message_register: Arc<MessageRegister>,
    reconnect_interval: Duration,
    stream_addr: Option<Addr<TcpStreamActor>>,
}

impl TcpConnecterActor {
    pub fn new(
        actor_name: &str,
        node_type: &str,
        node_name: &str,
        remote_addr: &str,
        reconnect_interval: Duration,
        message_register: Arc<MessageRegister>,
    ) -> Self {
        Self {
            self_addr: None,
            actor_name: actor_name.into(),
            node_type: node_type.into(),
            node_name: node_name.into(),
            remote_addr: remote_addr.into(),
            reconnect_interval,
            message_register,
            stream_addr: None,
        }
    }
}

impl std::fmt::Display for TcpConnecterActor {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        write!(f, "{}", self.name())
    }
}

#[async_trait::async_trait]
impl Actor for TcpConnecterActor {
    fn name(&self) -> &str {
        &self.actor_name
    }

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        let addr = ctx.address();
        self.self_addr = Some(addr.clone());
        ctx.register_center(&self.actor_name).await?;
        addr.send(Connect).ok();
        Ok(())
    }

    async fn stopped(&mut self, ctx: &mut Context<Self>) {
        if let Err(err) = ctx.unregister_center().await {
            error!("{} unregister_center failed:{}", self, err);
        }
        info!("{} stoped", self);
    }
}

#[message]
struct Connect;
#[async_trait::async_trait]
impl Handler<Connect> for TcpConnecterActor {
    async fn handle(&mut self, ctx: &mut Context<Self>, _: Connect) -> Result<()> {
        if self.stream_addr.is_some() {
            return Err(anyhow!("already connected"));
        }
        info!("try connect to {}@{} ..", self, self.remote_addr);
        match TcpStream::connect(&self.remote_addr).await {
            Ok(stream) => {
                stream.set_nodelay(true)?;
                let mut framed_write =
                    FramedWrite::new(stream.clone().compat(), HoneybeeMsgCodec::new());
                let mut framed_read =
                    FramedRead::new(stream.clone().compat(), HoneybeeMsgCodec::new());
                framed_write
                    .send_ext(TcpRegister::new(&self.node_type, &self.node_name))
                    .await?;

                let msg: TcpRegister = framed_read.read_msg().await?;

                let remote_node_type = msg.node_type;
                let remote_node_name = msg.node_name;

                let stream_addr = TcpStreamActor::new(
                    &remote_node_type,
                    &remote_node_name,
                    framed_write,
                    framed_read,
                    self.message_register.clone(),
                    None,
                )
                .start()
                .await?;

                self.stream_addr = Some(stream_addr.clone());
                info!(
                    "{}({}.{}@{}) connected",
                    self, remote_node_type, remote_node_name, self.remote_addr
                );

                let addr = ctx.address();
                let reconnect_interval = self.reconnect_interval;
                spawn(async move {
                    stream_addr.wait_for_stop().await;
                    sleep(reconnect_interval).await;
                    let _ = addr.send(DisConnected);
                });
            }
            Err(err) => {
                error!(
                    "connect to {}@{} failed err:{}, try reconnect after {}s",
                    self,
                    self.remote_addr,
                    err,
                    self.reconnect_interval.as_secs(),
                );
                ctx.send_later(Connect, self.reconnect_interval);
            }
        }
        Ok(())
    }
}

#[message]
struct DisConnected;
#[async_trait::async_trait]
impl Handler<DisConnected> for TcpConnecterActor {
    async fn handle(&mut self, ctx: &mut Context<Self>, _: DisConnected) -> Result<()> {
        error!("{} disconnected", self);
        self.stream_addr = None;
        ctx.address().send(Connect)
    }
}

pub struct SendMsg<T: Message<Result = ()> + HoneybeeMessage>(pub T);
impl<T: Message<Result = ()> + HoneybeeMessage> Message for SendMsg<T> {
    type Result = ();
}
#[async_trait::async_trait]
impl<T> Handler<SendMsg<T>> for TcpConnecterActor
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: SendMsg<T>) -> Result<()> {
        if let Some(ref stream) = self.stream_addr {
            if let Err(err) = stream.send_remote(msg.0) {
                error!(
                    "{} send msg:{} failed, {}, err:{}",
                    self,
                    T::ID,
                    self.remote_addr,
                    err
                );
            }
        } else {
            error!(
                "{} send msg:{:?} failed, {} disconnected",
                self, msg.0, self.remote_addr
            );
        }
        Ok(())
    }
}

pub struct CallMsg<T: Message<Result = R> + HoneybeeMessage, R: RpcResult>(
    pub oneshot::Sender<std::result::Result<R, RpcResponseError>>,
    pub T,
);
impl<T: Message<Result = R> + HoneybeeMessage, R: RpcResult> Message for CallMsg<T, R> {
    type Result = ();
}
#[async_trait::async_trait]
impl<T, R> Handler<CallMsg<T, R>> for TcpConnecterActor
where
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: CallMsg<T, R>) -> Result<()> {
        if let Some(ref stream) = self.stream_addr {
            let _ = stream.send(super::tcpstream::CallMsg(msg.0, msg.1));
        } else {
            error!(
                "{} call msg:{:?} failed, {}, not connected",
                self, msg.1, self.remote_addr
            );
        }
        Ok(())
    }
}

pub struct AsyncCallMsg<
    A: Actor + AnyHandler<R>,
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
>(pub Addr<A>, pub T);

impl<A: Actor + AnyHandler<R>, T: Message<Result = R> + HoneybeeMessage, R: RpcResult> Message
    for AsyncCallMsg<A, T, R>
{
    type Result = ();
}

#[async_trait::async_trait]
impl<A, T, R> Handler<AsyncCallMsg<A, T, R>> for TcpConnecterActor
where
    A: Actor + AnyHandler<R>,
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: AsyncCallMsg<A, T, R>) -> Result<()> {
        if let Some(ref stream) = self.stream_addr {
            let _ = stream.send(super::tcpstream::AsyncCallMsg(msg.0, msg.1));
        } else {
            error!(
                "{} async call msg:{:?} failed, {} disconnected",
                self, msg.1, self.remote_addr
            );
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl RemoteAddrExt for Addr<TcpConnecterActor> {
    fn send_remote<T>(&self, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage,
    {
        self.send(SendMsg(msg))
    }

    async fn call_remote<T, R>(&self, msg: T) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        let (tx, rx) = oneshot::channel();

        self.send(CallMsg(tx, msg))
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?;

        rx.await
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?
    }

    async fn call_remote_timeout<T, R>(
        &self,
        msg: T,
        dur: Duration,
    ) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        let (tx, rx) = oneshot::channel();

        self.send(CallMsg(tx, msg))
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?;

        timeout(dur, rx)
            .await
            .map_err(|_| RpcResponseError::RpcTimeOut(dur.as_millis()))?
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?
    }

    fn async_call_remote<A, T, R>(&self, source: Addr<A>, msg: T) -> Result<()>
    where
        A: Actor + AnyHandler<R>,
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        self.send(AsyncCallMsg(source, msg))
    }
}

impl UserSender for Addr<TcpConnecterActor> {
    fn send_to_user<T>(&self, uid: i64, msg: T)
    where
        T: 'static + prost::Message + MessageId,
    {
        let bytes = {
            let mut buf = Vec::with_capacity(msg.encoded_len());
            if let Err(err) = msg.encode(&mut buf) {
                error!("uid:{} send msg:{} err:{}", uid, T::ID, err);
                return;
            }
            buf
        };
        if let Err(err) = self.send_remote(UserMessageForward {
            uid,
            msg_id: T::ID,
            bytes,
        }) {
            error!("uid:{} send msg:{} err:{}", uid, T::ID, err);
        }
    }
}

impl AppUserSender for Addr<TcpConnecterActor> {
    fn send_to_app_user<T>(&self, uid: i64, msg: T)
    where
        T: Message<Result = ()> + HoneybeeMessage,
    {
        match bincode::serialize(&msg) {
            Ok(bytes) => {
                if let Err(err) = self.send_remote(UserMessageApp {
                    uid,
                    msg_id: T::ID,
                    bytes,
                }) {
                    error!("send_remote uid:{} msg:{:?} failed:{}", uid, msg, err);
                }
            }
            Err(err) => error!("send_to_app_user uid:{} msg:{:?} failed:{}", uid, msg, err),
        }
    }
}
